package org.pengge.rxjava.web.controller.first;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import lombok.extern.slf4j.Slf4j;
import org.pengge.rxjava.web.entity.vo.ResponseVO;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 * @author
 * @since 2017/7/11 下午7:26
 * Purpose
 */
@Controller
@RequestMapping("/first/chapter1")
@Slf4j
public class ChapterOneController {
    /**
     * 第一章第一个案例
     *
     * @return
     */
    @RequestMapping("/case1")
    @ResponseBody
    public ResponseVO case1() {
        ResponseVO responseVO = new ResponseVO();
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //注意针对web项目，必须要写这个结束语句，不然会一直等待
                emitter.onComplete();
            }
        });
        //创建一个下游 Observer
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                log.info("subscribe");
            }

            @Override
            public void onNext(Integer value) {
                log.info("" + value);
            }

            @Override
            public void onError(Throwable e) {
                log.info("error");
            }

            @Override
            public void onComplete() {
                log.info("complete");
            }
        };
        //建立连接
        observable.subscribe(observer);
        //blockingIterable转换Observable为其他对象的方法
        Iterable<Integer> integers = observable.blockingIterable();
        responseVO.setData(integers);
        return responseVO;
    }

    /**
     * 主要展示链式操作符
     *
     * @return
     */
    @RequestMapping("/case2")
    @ResponseBody
    public ResponseVO case2() {
        ResponseVO responseVO = new ResponseVO();
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                log.info("subscribe");
            }

            @Override
            public void onNext(Integer value) {
                log.info("" + value);
            }

            @Override
            public void onError(Throwable e) {
                log.info("error");
            }

            @Override
            public void onComplete() {
                log.info("complete");
            }
        });

        Iterable<Integer> integers = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).blockingIterable();

        responseVO.setData(integers);
        return responseVO;
    }

    /**
     * 测试Disposable
     * @return
     */
    @RequestMapping("/case3")
    @ResponseBody
    public ResponseVO case3() {
        ResponseVO responseVO = new ResponseVO();
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                log.info("emit 1");
                emitter.onNext(1);
                log.info("emit 2");
                emitter.onNext(2);
                log.info("emit 3");
                emitter.onNext(3);
                log.info("emit complete");
                emitter.onComplete();
                log.info("emit 4");
                emitter.onNext(4);
            }
        });
        observable.subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                log.info("subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                log.info("onNext: " + value);
                i++;
                if (i == 2) {
                    log.info("dispose");
                    mDisposable.dispose();
                    log.info("isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                log.info("error");
            }

            @Override
            public void onComplete() {
                log.info("complete");
            }
        });
        responseVO.setData(observable.blockingIterable());
        return responseVO;
    }
}
